﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;
using Newtonsoft.Json;
using System.Collections.Concurrent;
using System.Threading.Actors.Languages;
using System.Threading.Actors.Utilities;
using System.Diagnostics.Contracts;

namespace System.Threading.Actors
{
    // Listens for ActorServiceRequests, responds with ActorServiceResults
    public class ActorRawTcpListener
    {
        private TcpListener m_listener;
        private bool m_shutdown = false;  // Note we currently have no shutdown logic for listeners, because they are for ports.

        // Allows for multiple "global" listeners with different ports.
        // Even though I don't initially expect the need for this.
        private static readonly Dictionary<int, ActorRawTcpListener> s_listenerMap = new Dictionary<int, ActorRawTcpListener>();

        // Map "discriminator strings" (i.e., the "path" tacked onto the Uri) to actors.
        internal readonly ConcurrentDictionary<string, IActor> m_actorMap = new ConcurrentDictionary<string, IActor>();

        // Map discriminator strings to contexts, so that we can dispose contexts when actors are demoted to secondaries.
        private readonly Dictionary<string, HashSet<ClientTcpContext>> m_contextMap = new Dictionary<string, HashSet<ClientTcpContext>>();

        // Gets the global listener for the indicated port.
        public static ActorRawTcpListener GetListener(Uri uri)
        {
            Contract.Requires(uri != null);

            int port = uri.Port;
            ActorRawTcpListener result;
            if (!s_listenerMap.TryGetValue(port, out result))
            {
                // If this lock becomes a problem, consider replacing it with a SemaphoreSlim.
                lock (s_listenerMap)
                {
                    if (!s_listenerMap.TryGetValue(port, out result))
                    {
                        result = new ActorRawTcpListener(uri);
                        Thread.MemoryBarrier();
                        s_listenerMap[port] = result;
                    }
                }
            }
            return result;
        }

        private ActorRawTcpListener(Uri uri)
        {
            Console.WriteLine("{0}: ActorRawTcpListener ctor({1}) called", DateTime.Now, uri.ToString());
            var ipAddress = NetworkConfiguration.GetIPAddress(NetworkConfiguration.PreferIPv6 ? AddressFamily.InterNetworkV6 : AddressFamily.InterNetwork);
            IPEndPoint ep = new IPEndPoint(ipAddress, uri.Port);
            m_listener = new TcpListener(ep);
            m_listener.Start();
            // ListenerLoop almost immediately does an await, and is called from within a lock.
            // We need to escape it to another thread.
            Task.Run(() => 
            { 
                ListenerLoop(); // fire-and-forget 
            } );
            this.Address = uri.ToString();

        }

        // Registers an actor as a target for the ActorRawTcpListener, using the uri localPath
        // as a discriminator.
        public void RegisterService(Uri uri, IActor service)
        {
            Contract.Requires(uri != null);
			
            var localPath = uri.LocalPath;
            m_actorMap[localPath] = service;
        }

        // Deregisters a localPath discriminator; called when a service is closed or demoted
        public void UnregisterService(Uri uri)
        {
            if (uri == null) return;
            IActor unused;
            var discriminator = uri.LocalPath;
            m_actorMap.TryRemove(discriminator, out unused);
            // Dispose all contexts associated with this discriminator
            List<ClientTcpContext> contextsToBeDisposed = null;
            lock (m_contextMap)
            {
                HashSet<ClientTcpContext> set;
                if (m_contextMap.TryGetValue(discriminator, out set))
                {
                    // Make a copy of client contexts to be disposed, so that we can do it outside the lock
                    contextsToBeDisposed = new List<ClientTcpContext>(set);
                    // Remove any record of this discriminator
                    m_contextMap.Remove(discriminator);
                }
            }

            if (contextsToBeDisposed != null)
            {
                foreach (var context in contextsToBeDisposed)
                {
                    // Close the context, eating any ensuing exceptions
                    try { context.Dispose(); } // This will result in a call to RecordContextDeath, but that will be a no-op since we've removed the discriminator as a key
                    catch { }
                }
            }

        }

        // Called by spawned contexts; maps a discriminator to a set of contexts
        internal void RegisterContext(ClientTcpContext context, string discriminator)
        {
            lock (m_contextMap)
            {
                HashSet<ClientTcpContext> set;
                if (!m_contextMap.TryGetValue(discriminator, out set))
                {
                    set = new HashSet<ClientTcpContext>();
                    m_contextMap[discriminator] = set;
                }

                set.Add(context);
            }
        }

        // Called from a spawned context's Dispose() method
        internal void RecordContextDeath(ClientTcpContext context, string discriminator)
        {
            lock (m_contextMap)
            {
                HashSet<ClientTcpContext> set;
                if (m_contextMap.TryGetValue(discriminator, out set))
                {
                    set.Remove(context);
                }
            }
        }

        public string Address { get; private set; }

        // Loops waiting for incoming connection requests.
        private async void ListenerLoop()
        {
            while (!m_shutdown)
            {
                try
                {
                    var client = await m_listener.AcceptTcpClientAsync();
                    var context = new ClientTcpContext(client, this);
                }
                catch (SocketException)  // thrown by ClientTcpContext's ClientLoop, if client dies quickly.
                {
                }
                catch (Exception e)
                {
                    // listener must have been shut down.  Presumably, m_shutdown is
                    // set and our ListenerLoop will exit.
                    Contract.Assert(m_shutdown, "Unexpected error in ActorRawTcpListener::ListenerLoop.  " + e);
                }
            }
        }

    }

    // Holds information pertinent to an individual client
    internal class ClientTcpContext : IPublicationCallbackContract, IDisposable
    {
        private readonly TcpClient m_client;
        private readonly NetworkStream m_clientStream;
        private volatile bool m_disposed = false;
        private readonly ActorRawTcpListener m_parent;
        private string m_discriminator;
        private SemaphoreSlim m_writeSemaphore; // Need to lock because we're multiplexing with responses and other events

        // Constructor.  Requires an underlying TcpClient and a reference to an Actor service.
        public ClientTcpContext(TcpClient client, ActorRawTcpListener parent)
        {
            m_parent = parent;
            m_client = client;
            m_client.NoDelay = true; // Eliminate delayed transfer for writes
            m_clientStream = client.GetStream();
            m_writeSemaphore = new SemaphoreSlim(1, 1);
            ClientLoop();  // Fire and forget
            //Console.WriteLine("ClientTcpContext started for client {0}", client.Client.RemoteEndPoint.ToString());
        }

        // If this throws, it will alert the publish logic that this client needs to be removed from
        // the subscription list.
        public void OnNewEvent(ActorEvent actorEvent)
        {
            //Console.WriteLine("Attempting to publish event {0} to {1}", actorEvent.EventType, m_client.Client.RemoteEndPoint.ToString());
            try
            {
                // Two problems with using async writes here:
                // 1) This method is synchronous, so we would have to block anyways.
                // 2) Not yet convinced the async methods are safe, though I couldn't find an obvious bug in
                //    Stream's WriteAsync or the TPL code it relies upon.
                /*
                var task = SerializationHelper.SerializeActorEventAsync(actorEvent, m_clientStream, m_writeSemaphore);
                task.Wait();
                 */
                SerializationHelper.SerializeActorEvent(actorEvent, m_clientStream, m_writeSemaphore);
            }
            catch { Dispose(); throw; }
        }

        
        public void OnCompleting()
        {
        }

        public void Dispose()
        {
            if (!m_disposed)
            {
                m_disposed = true;
                m_clientStream.Close();
                m_client.Close();
                m_writeSemaphore.Dispose();

                // Remove any record of yourself from the ActorRawTcpListener
                if (m_discriminator != null)
                {
                    m_parent.RecordContextDeath(this, m_discriminator);
                }
            }
        }

        // Loop on client input
        private async void ClientLoop()
        {
            var messageBuf = new byte[1024];
            EndPoint clientEndPoint = m_client.Client.RemoteEndPoint;  // Mostly for debugging
            try
            {
                // First thing: receive the discriminator string from the client.
                m_discriminator = await Utilities.NetworkHelpers.ReadStringAsync(m_clientStream, messageBuf).ConfigureAwait(continueOnCapturedContext: false);
                string partitionName = null;

                // Retrieve partition key and discriminator
                int hashIndex = m_discriminator.IndexOf('#');
                if (hashIndex >= 0)
                {
                    if (hashIndex + 1 < m_discriminator.Length)
                    {
                        partitionName = m_discriminator.Substring(hashIndex + 1);
                    }
                    m_discriminator = m_discriminator.Substring(0, hashIndex);
                }

                IActor targetActor;
                if (!m_parent.m_actorMap.TryGetValue(m_discriminator, out targetActor))
                {
                    // There is no actor associated with this discriminator.  Reply with an error.
                    var errorMsg = String.Format("EXCEPTION:ActorNotFoundException:No actor found for discriminator {0}", m_discriminator);
                    Utilities.NetworkHelpers.WriteString(errorMsg, m_clientStream);
                    Dispose();
                    return;
                }

                m_parent.RegisterContext(this, m_discriminator);

                // We successfully got the actor from the discriminator.  Send an "OK".
                Utilities.NetworkHelpers.WriteString("OK", m_clientStream);

                // Now enter the loop
                while (!m_disposed)
                {
                    var req = await SerializationHelper.DeserializeActorServiceRequestAsync(m_clientStream, messageBuf).ConfigureAwait(continueOnCapturedContext: false);

                    await EvalActorServiceRequestAsync(req, targetActor, partitionName, m_clientStream, m_writeSemaphore, this).ConfigureAwait(continueOnCapturedContext: false);
                }
                //Console.WriteLine("ActorRawTcpListener::ClientLoop done - we disposed this connection.");
            }
            catch (IOException)
            {
                //Console.WriteLine("ActorRawTcpListener::ClientLoop - Client {0} disconnected", clientEndPoint.ToString());
                Dispose();
            }
            catch (ObjectDisposedException)
            {
                //Console.WriteLine("ActorRawTcpListener::ClientLoop - caught an ObjectDisposedException.  Client was {0}", clientEndPoint.ToString());
                // It's possible that an async channel close could cause an ODE here
                Dispose(); // Probably unnecessary
            }
            catch (Exception e)
            {
                Console.WriteLine("ActorRawTcpListener::ClientLoop hit an unexpected exception and is crashing!  {0}", e);
                Contract.Assert(false, String.Format("Unhandled exception in ActorRawTcpListener::ClientLoop.\r\n{0}", e));
            }
        }

        // Note: This async method returns a Task.  The compiler will wire up the code such that awaiting this method will await
        // not just any inner tasks here (such as from calling CallMethod), but will also not complete until the continuation is 
        // complete, meaning we will wait until the response has been written.
        // While this could theoretically be an instance method on ClientTcpContext, if we support multiplexing in the future
        // we must avoid any bugs where we use the wrong ClientTcpContext instance for responding to the stream, etc.  For this
        // reason, I'm making this static, even though we need a ClientTcpContext instance passed in for subscribe & unsubscribe methods.
        private static async Task EvalActorServiceRequestAsync(ActorServiceRequest req, IActor targetActor, String partitionName,
            Stream responseStream, SemaphoreSlim writeSemaphore, ClientTcpContext tcpContextForPubSub)
        {
            //Console.WriteLine("Client got request \"{0}\" from {1}", req.MethodName, tcpContextForPubSub.m_client.Client.RemoteEndPoint.ToString());
            string asrString;
            try
            {
                if (req.MethodName.Equals(ActorOperations.CALLMETHOD))
                {
                    // If we don't deserialize the method name, it will have quotes around it.
                    var methodName = (string)SerializationHelper.Deserialize(req.MethodArgs[0]);
                    // These can remain JSON-encoded
                    var methodArgs = req.MethodArgs.Skip(1).ToArray();
                    asrString = await targetActor.CallMethod(req.UniqueClientId, req.ClientSequenceNumber, partitionName, methodName, methodArgs);
                }
                else if (req.MethodName.Equals(ActorOperations.ADDASSEMBLY))
                {
                    var assemblyName = (string)SerializationHelper.Deserialize(req.MethodArgs[0]);
                    // Json.Net implicitly serializes byte arrays as Base64 strings.  We'll need to
                    // explicitly convert it back.
                    var assemblyBytesString = (string)SerializationHelper.Deserialize(req.MethodArgs[1]);
                    var assemblyBytes = Convert.FromBase64String(assemblyBytesString);
                    asrString = await targetActor.AddAssembly(req.ClientSequenceNumber, partitionName, assemblyName, assemblyBytes);
                }
                else if (req.MethodName.Equals(ActorOperations.DELETEACTOR))
                {
                    asrString = targetActor.DeleteSelf(req.ClientSequenceNumber, partitionName);
                }
                else if (req.MethodName.Equals(ActorOperations.SUBSCRIBE))
                {
                    asrString = targetActor.Subscribe(req.ClientSequenceNumber, partitionName,
                        (string)SerializationHelper.Deserialize(req.MethodArgs[0]), tcpContextForPubSub);
                }
                else if (req.MethodName.Equals(ActorOperations.UNSUBSCRIBE))
                {
                    asrString = targetActor.Unsubscribe(req.ClientSequenceNumber, partitionName,
                        (string)SerializationHelper.Deserialize(req.MethodArgs[0]), tcpContextForPubSub);
                }
                else if (req.MethodName.Equals(ActorOperations.UNSUBSCRIBEALL))
                {
                    asrString = targetActor.UnsubscribeAll(req.ClientSequenceNumber, partitionName, tcpContextForPubSub);
                }
                else if (req.MethodName.Equals(ActorOperations.PING))
                {
                    //Trace.WriteLine(String.Format("Actor responding to ping.  ID: {0}  Client Seq #: {1} part: {2}  managed Thread ID: {3}  PID: {4}", req.UniqueClientId, req.ClientSequenceNumber, partitionName, Environment.CurrentManagedThreadId, Process.GetCurrentProcess().Id));
                    asrString = targetActor.Ping(req.ClientSequenceNumber, partitionName);
                }
                else if (req.MethodName.Equals(ActorOperations.CONFIGURE))
                {
                    asrString = targetActor.Configure(req.ClientSequenceNumber, partitionName, (string)SerializationHelper.Deserialize(req.MethodArgs[0]));
                }
                else if (req.MethodName.Equals(ActorOperations.ADDMETHOD))
                {
                    var language = (SupportedLanguages)(int)SerializationHelper.Deserialize(req.MethodArgs[0]);
                    var methodName = (string)SerializationHelper.Deserialize(req.MethodArgs[1]);
                    var methodImpl = (string)SerializationHelper.Deserialize(req.MethodArgs[2]);
                    asrString = await targetActor.AddMethod(req.ClientSequenceNumber, partitionName, (int)language, methodName, methodImpl);
                }
                else throw new InvalidOperationException("Unrecognized actor service operation \"" + req.MethodName + "\"");
            }
            catch (Exception e)
            {
                asrString = SerializationHelper.Serialize(ActorServiceResult.FromException(req.ClientSequenceNumber, e));
            }

            // If this wasn't a fire & forget request, then send a response back to the client.
            if (req.ClientSequenceNumber != ActorServiceRequest.FireAndForgetSequenceNumber) // "No Response Expected"
            {
                // BUGBUG: Tried using async writes here, but that doesn't appear to be safe.
                // Either we're writing to that stream without taking an appropriate lock somewhere, or
                // there's a bug where this lets us call NetworkStream's BeginWrite in a non-threadsafe manner.
                // Looks like we're writing half a response array then the entire response again.
                // I couldn't find an obvious bug in Stream's WriteAsync or the TPL code it relies upon.
                //await SerializationHelper.SerializeActorServiceResultAsync(asrString, responseStream, writeSemaphore);
                SerializationHelper.SerializeActorServiceResult(asrString, responseStream, writeSemaphore);
#if DEBUG
                //Socket socket = tcpContextForPubSub.m_client.Client;
                //Trace.WriteLine(String.Format("Client was sent result \"{0}\" to {1}.  Local endpoint: {2}", asrString, 
                //    socket.RemoteEndPoint.ToString(), socket.LocalEndPoint.ToString()));
#endif
            }
        }
    }
}
